Add provisional per-subject evaluation telemetry opt-out tests - #8080

diff --git a/manifests/c.yml b/manifests/c.yml
index 23631e4c4..38176847a 100644
--- a/manifests/c.yml
+++ b/manifests/c.yml
@@ -4,7 +4,7 @@ manifest:
   tests/ai_guard/: irrelevant (dd-trace-c does not implement AI Guard)
   tests/appsec/: irrelevant (dd-trace-c does not implement AppSec or IAST)
   tests/cws/test_thread_context_sharing.py::Test_ThreadContextSharing: missing_feature (missing /security/thread_context_sharing endpoint on weblog)
-  tests/ffe/test_evaluation_telemetry_opt_out.py: missing_feature (Provisional per-subject evaluation telemetry opt-out coverage; enable after the SDK release)
+  tests/ffe/test_evaluation_telemetry_opt_out.py: incomplete_test_app (The Perl Mojolicious workload does not provide the /ffe endpoint)
   tests/integrations/test_dbm.py: incomplete_test_app (The Perl Mojolicious workload does not provide database endpoints)
   tests/parametric/otel_env_vars/test_otel_exporter_otlp_compression.py::Test_OTEL_EXPORTER_OTLP_COMPRESSION: missing_feature
   tests/parametric/otel_env_vars/test_otel_exporter_otlp_headers.py::Test_OTEL_EXPORTER_OTLP_HEADERS: missing_feature
diff --git a/manifests/cpp.yml b/manifests/cpp.yml
index 113a08ad8..99b11c7e9 100644
--- a/manifests/cpp.yml
+++ b/manifests/cpp.yml
@@ -22,7 +22,7 @@ manifest:
   tests/debugger/test_debugger_exception_replay.py::Test_Debugger_Exception_Replay::test_exception_replay_stackoverflow: missing_feature (Implemented only for dotnet)
   tests/debugger/test_debugger_probe_snapshot.py::Test_Debugger_Line_Probe_Snaphots::test_process_tags_snapshot: missing_feature (Not yet implemented)
   tests/debugger/test_debugger_probe_snapshot.py::Test_Debugger_Line_Probe_Snaphots::test_process_tags_snapshot_svc: missing_feature (Not yet implemented)
-  tests/ffe/test_evaluation_telemetry_opt_out.py: missing_feature (Provisional per-subject evaluation telemetry opt-out coverage; enable after the SDK release)
+  tests/ffe/test_evaluation_telemetry_opt_out.py: irrelevant (only parametric tests are run for cpp)
   tests/ffe/test_flag_eval_evp.py: irrelevant (only parametric tests are run for cpp)
   tests/integration_frameworks/llm/anthropic/test_anthropic_llmobs.py::TestAnthropicLlmObsMessages::test_create_error: bug (MLOB-1234)
   tests/integrations/crossed_integrations/test_sqs.py::Test_SQS_PROPAGATION_VIA_AWS_XRAY_HEADERS: irrelevant (Localstack SQS does not support AWS Xray Header parsing)
diff --git a/manifests/dotnet.yml b/manifests/dotnet.yml
index 7075d94ea..f86b80dde 100644
--- a/manifests/dotnet.yml
+++ b/manifests/dotnet.yml
@@ -823,7 +823,7 @@ manifest:
     - declaration: bug (FFL-2184)
       excluded_weblog: [poc, uds]
   tests/ffe/test_evaluation_telemetry_opt_out.py: missing_feature (Provisional per-subject evaluation telemetry opt-out coverage; enable after the SDK release)
-  tests/ffe/test_evaluation_telemetry_opt_out.py::Test_FFE_Evaluation_Telemetry_Opt_Out_EVP: missing_feature (.NET has no EVP evaluation producer; FFL-2446)
+  tests/ffe/test_evaluation_telemetry_opt_out.py::Test_FFE_Evaluation_Telemetry_Opt_Out_EVP: missing_feature (.NET has no EVP evaluation producer until dd-trace-dotnet#9351; FFL-2446)
   tests/ffe/test_exposure_egress.py: v3.36.0
   tests/ffe/test_exposure_egress.py::Test_FFE_Exposure_Egress_Agentless_Direct: missing_feature (Not yet implemented)
   tests/ffe/test_exposure_egress.py::Test_FFE_Exposure_Egress_Agentless_Direct_Shutdown: missing_feature (Not yet implemented)
diff --git a/tests/ffe/README.md b/tests/ffe/README.md
index 701477696..76ab37709 100644
--- a/tests/ffe/README.md
+++ b/tests/ffe/README.md
@@ -26,6 +26,7 @@ All four test classes report under [#649: Per-subject evaluation telemetry opt-o
 Candidate SDK implementations must pass the applicable local checks before delivery.
 Released-version CI activation is a separate step. Both scenarios are excluded from the default
 `all`, `end_to_end`, and `tracer_release` groups. They remain selectable by name or through the `ffe` group.
+When a manifest activates these tests, restore the default groups. Otherwise tracer CI that selects only `tracer_release` does not run them.
 Manifests mark unsupported SDK versions as expected failures; this does not prevent their setup from running.
 Use a local candidate SDK and force these tests to validate unreleased changes:
 
@@ -42,7 +43,10 @@ The disabled scenario leaves the setting absent. Both use the same requests.
 | Context value | Enabled | Disabled |
 | --- | --- | --- |
 | Boolean `true` | No evaluation telemetry | Evaluation telemetry remains |
-| Boolean `false`, missing, string `"true"`, or number `1` | Evaluation telemetry remains | Evaluation telemetry remains |
+| Boolean `false`, missing, `null`, string `"true"`, or number `1` | Evaluation telemetry remains | Evaluation telemetry remains |
+
+The SDK reads the attribute on every evaluation. A subject stays opted out only while each evaluation context carries it.
+Set the attribute in the evaluation context supplied to the call. These tests do not cover opt-outs added by application hooks, and SDKs handle those differently.
 
 Separate tests check exact EVP counts, evaluation metric presence, and automatic trace annotations.
 Metric checks accept delta exports and repeated cumulative exports. Opted-out flags must have no nonzero metric points.
@@ -51,13 +55,27 @@ Mixed subjects evaluate the same flag to detect a flag-scoped opt-out decision.
 EVP counts match both the flag and the subject's raw or hashed targeting key.
 Shared-flag subjects receive the same value from distinct fixture allocations.
 Metric assertions use the allocation tags to check each subject separately, without adding subject tags to metrics.
+One subject opts out and then evaluates without the attribute, to detect an opt-out that sticks to the subject.
+Flags absent from the configuration check the error path, where evaluations return the default value.
 One rule targets the opt-out attribute to verify that it remains available for evaluation.
-All requests must return the same flag value. Exposures must retain their ordinary deduplication and context attributes.
+Configured flags must return the same flag value for every request.
+Exposures must keep their context attributes and their ordinary deduplication for an unchanged context.
+After the attribute value changes, exposure deduplication can differ between SDKs, and the tests do not check it.
 Validation reads final captures after the scenario stops its application and drains the Agent.
 Remote Configuration and evaluation-request errors are reported during validation, without aborting the remaining setups.
 
 PHP coverage is deferred until its OpenFeature test-app foundation is available.
-.NET has no EVP evaluation producer. Exclude the EVP class when forcing its local runs with `-k 'not EVP'`.
+.NET has no EVP evaluation producer until DataDog/dd-trace-dotnet#9351. Exclude the EVP class when forcing its local runs with `-k 'not EVP'`.
+For a local .NET candidate, put the `Datadog.Trace` and `Datadog.FeatureFlags.OpenFeature` packages and the matching
+tracer (a `datadog-dotnet-apm-<version>*.tar.gz` or extracted tracer files) in `binaries/`. Pin the package versions with brackets,
+so that restore fails instead of selecting another version (`poc` weblog only):
+
+```bash
+./build.sh dotnet -w poc -e "--build-arg DD_TRACE_PACKAGE_VERSION=[<version>] --build-arg DD_OPENFEATURE_PACKAGE_VERSION=[<version>]"
+```
+
+A version without brackets is a minimum. If that version is missing and a higher one is available, NuGet restores the higher one and only warns (NU1603).
+Use candidate versions that are not on nuget.org, so that only `binaries/` can supply them.
 Browser and mobile SDKs are outside this server harness and retain their repository integration tests.
 
 ## Test-contract self-tests
diff --git a/tests/ffe/test_evaluation_telemetry_opt_out.py b/tests/ffe/test_evaluation_telemetry_opt_out.py
index 94de09b5c..d1df617ff 100644
--- a/tests/ffe/test_evaluation_telemetry_opt_out.py
+++ b/tests/ffe/test_evaluation_telemetry_opt_out.py
@@ -30,6 +30,7 @@ ATTRIBUTES: dict[str, dict[str, Any]] = {
     "missing": {},
     "string-true": {ATTRIBUTE: "true"},
     "number-one": {ATTRIBUTE: 1},
+    "null": {ATTRIBUTE: None},
 }
 ANNOTATIONS = {"ffe_flags_enc", "ffe_subjects_enc", "ffe_runtime_defaults"}
 
@@ -37,9 +38,11 @@ ANNOTATIONS = {"ffe_flags_enc", "ffe_subjects_enc", "ffe_runtime_defaults"}
 class EvaluationCase(TypedDict):
     flag: str
     subject: str
-    allocation_key: str
+    allocation_key: str | None
     attributes: dict[str, Any]
     expected_count: int
+    expected_value: str
+    check_exposure: bool
     responses: list[HttpResponse]
 
 
@@ -53,7 +56,7 @@ class _TelemetryContract:
         enabled = context.scenario == scenarios.feature_flagging_evaluation_telemetry_opt_out
         self.cases = []
         fixture = make_ufc_fixture(f"opt-out-{signal}-true", serial_id=101)
-        fixture["flags"] = {}
+        fixture["flags"] = {}  # Keep only the envelope; flags are added below.
         for index, (label, attributes) in enumerate(ATTRIBUTES.items()):
             flag = f"opt-out-{signal}-{label}"
             fixture["flags"].update(make_ufc_fixture(flag, serial_id=101 + index)["flags"])
@@ -76,10 +79,28 @@ class _TelemetryContract:
                     allocation_key="default-allocation",
                     attributes=attributes,
                     expected_count=0 if enabled and label == "true" else 3,
+                    expected_value="on-value",
+                    check_exposure=True,
                     responses=[],
                 )
             )
 
+        # Flags absent from the configuration take the error path and return the default.
+        # The error result carries no allocation, so metrics are matched by flag only.
+        self.cases.extend(
+            EvaluationCase(
+                flag=f"opt-out-{signal}-missing-flag-{label}",
+                subject=f"subject-{signal}-missing-flag-{label}",
+                allocation_key=None,
+                attributes={ATTRIBUTE: opted_out},
+                expected_count=0 if enabled and opted_out else 3,
+                expected_value="default",
+                check_exposure=False,
+                responses=[],
+            )
+            for label, opted_out in [("true", True), ("false", False)]
+        )
+
         independent_cases = list(self.cases)
         shared_flag = f"opt-out-{signal}-shared"
         fixture["flags"].update(make_ufc_fixture(shared_flag)["flags"])
@@ -91,6 +112,8 @@ class _TelemetryContract:
                 allocation_key=f"shared-{label}",
                 attributes={ATTRIBUTE: opted_out, "test_subject": f"subject-{signal}-shared-{label}"},
                 expected_count=0 if enabled and opted_out else 3,
+                expected_value="on-value",
+                check_exposure=True,
                 responses=[],
             )
             for label, opted_out in [("true", True), ("false", False)]
@@ -98,15 +121,37 @@ class _TelemetryContract:
         # Metrics do not contain subject identifiers. Distinct allocations let
         # their existing tags identify each subject without changing the value.
         for index, case in enumerate(shared_cases):
-            allocation = make_ufc_fixture(shared_flag, allocation_key=case["allocation_key"], serial_id=106 + index)[
-                "flags"
-            ][shared_flag]["allocations"][0]
+            allocation_key = case["allocation_key"]
+            assert allocation_key is not None
+            allocation = make_ufc_fixture(shared_flag, allocation_key=allocation_key, serial_id=110 + index)["flags"][
+                shared_flag
+            ]["allocations"][0]
             allocation["rules"] = [
                 {"conditions": [{"attribute": "test_subject", "operator": "ONE_OF", "value": [case["subject"]]}]}
             ]
             fixture["flags"][shared_flag]["allocations"].append(allocation)
         self.cases.extend(shared_cases)
 
+        # One subject opts out, then evaluates without the attribute. The opt-out is read
+        # per evaluation, so later evaluations must emit. Exposure deduplication after the
+        # attribute changes is SDK-defined, so exposures are not checked here.
+        per_evaluation_flag = f"opt-out-{signal}-per-evaluation"
+        fixture["flags"].update(make_ufc_fixture(per_evaluation_flag, serial_id=120)["flags"])
+        per_evaluation_cases = [
+            EvaluationCase(
+                flag=per_evaluation_flag,
+                subject=f"subject-{signal}-per-evaluation",
+                allocation_key="default-allocation",
+                attributes=attributes,
+                expected_count=0 if enabled and attributes else 3,
+                expected_value="on-value",
+                check_exposure=False,
+                responses=[],
+            )
+            for attributes in [{ATTRIBUTE: True}, {}]
+        ]
+        self.cases.extend(per_evaluation_cases)
+
         rc.tracer_rc_state.reset().set_config(f"datadog/2/FFE_FLAGS/opt-out-{signal}/config", fixture).apply()
 
         # Different subjects overlap. Repeated evaluations also exercise aggregation
@@ -134,10 +179,22 @@ class _TelemetryContract:
             for case in shared_cases:
                 evaluate(case)
 
+        for case in per_evaluation_cases:
+            for _ in range(3):
+                evaluate(case)
+
     def assert_evaluations(self) -> None:
         assert_setup_succeeded(self.setup_error)
         for case in self.cases:
-            assert_evaluation_responses(case["responses"], expected_count=3, expected_value="on-value")
+            assert_evaluation_responses(case["responses"], expected_count=3, expected_value=case["expected_value"])
+
+    def expected_totals(self, key: str) -> dict[tuple[str, str | None], int]:
+        # Cases that share a flag and a subject (or allocation) are checked together.
+        totals: dict[tuple[str, str | None], int] = {}
+        for case in self.cases:
+            group = (case["flag"], case["subject"] if key == "subject" else case["allocation_key"])
+            totals[group] = totals.get(group, 0) + case["expected_count"]
+        return totals
 
 
 @scenarios.feature_flagging_evaluation_telemetry_opt_out_disabled
@@ -151,12 +208,9 @@ class Test_FFE_Evaluation_Telemetry_Opt_Out_EVP(_TelemetryContract):
     def test_counts(self) -> None:
         self.assert_evaluations()
         captures = list(interfaces.agent.get_data(path_filters="/api/v2/flagevaluation"))
-        for case in self.cases:
-            assert_count(
-                evaluation_count(captures, case["flag"], subject=case["subject"]),
-                case["expected_count"],
-                f"{case['flag']} ({case['subject']})",
-            )
+        for (flag, subject), expected_count in self.expected_totals("subject").items():
+            assert subject is not None
+            assert_count(evaluation_count(captures, flag, subject=subject), expected_count, f"{flag} ({subject})")
 
 
 @scenarios.feature_flagging_evaluation_telemetry_opt_out_disabled
@@ -170,13 +224,8 @@ class Test_FFE_Evaluation_Telemetry_Opt_Out_Metrics(_TelemetryContract):
     def test_counts(self) -> None:
         self.assert_evaluations()
         metrics = [metric for _, metric in interfaces.agent.get_metrics()]
-        for case in self.cases:
-            assert_metric_activity(
-                metrics,
-                case["flag"],
-                allocation_key=case["allocation_key"],
-                suppressed=case["expected_count"] == 0,
-            )
+        for (flag, allocation_key), expected_count in self.expected_totals("allocation_key").items():
+            assert_metric_activity(metrics, flag, allocation_key=allocation_key, suppressed=expected_count == 0)
 
 
 @scenarios.feature_flagging_evaluation_telemetry_opt_out_disabled
@@ -191,6 +240,8 @@ class Test_FFE_Evaluation_Telemetry_Opt_Out_Exposures(_TelemetryContract):
         self.assert_evaluations()
         captures = list(interfaces.agent.get_data(path_filters="/api/v2/exposures"))
         for case in self.cases:
+            if not case["check_exposure"]:
+                continue
             events = [
                 event for data in captures for event in exposure_events_from_data(data, {case["flag"]}, case["subject"])
             ]
@@ -200,11 +251,14 @@ class Test_FFE_Evaluation_Telemetry_Opt_Out_Exposures(_TelemetryContract):
             attributes = events[0]["subject"].get("attributes", {})
             # Some providers also include targetingKey in exposure attributes.
             # Check the customer input without forbidding that existing behavior.
-            if ATTRIBUTE in case["attributes"]:
+            if ATTRIBUTE in case["attributes"] and case["attributes"][ATTRIBUTE] is None:
+                # SDKs may drop a null attribute or keep it as null.
+                assert attributes.get(ATTRIBUTE) is None, attributes
+            elif ATTRIBUTE in case["attributes"]:
                 expected = case["attributes"][ATTRIBUTE]
                 assert attributes[ATTRIBUTE] == expected
-                if isinstance(expected, bool):
-                    assert isinstance(attributes[ATTRIBUTE], bool)
+                # 1 == True in Python, so compare bool-ness in both directions.
+                assert isinstance(attributes[ATTRIBUTE], bool) == isinstance(expected, bool), attributes
             else:
                 assert ATTRIBUTE not in attributes
 
diff --git a/tests/ffe/test_evaluation_telemetry_opt_out_contract.py b/tests/ffe/test_evaluation_telemetry_opt_out_contract.py
index 7b753b66b..1248818c0 100644
--- a/tests/ffe/test_evaluation_telemetry_opt_out_contract.py
+++ b/tests/ffe/test_evaluation_telemetry_opt_out_contract.py
@@ -26,6 +26,8 @@ from utils import HttpResponse, features, pytest, scenarios
         (200, "not JSON", 3, False),
         (200, "{}", 3, False),
         (200, '{"value": "on-value"}', 0, False),
+        (200, '{"value": "on-value"}', 2, False),
+        (200, '{"value": "on-value"}', 4, False),
     ],
 )
 def test_evaluation_response_oracle(status_code: int, text: str, count: int, *, passes: bool) -> None:
diff --git a/tests/ffe/utils/evaluation_telemetry.py b/tests/ffe/utils/evaluation_telemetry.py
index fb4affe16..896913910 100644
--- a/tests/ffe/utils/evaluation_telemetry.py
+++ b/tests/ffe/utils/evaluation_telemetry.py
@@ -34,7 +34,7 @@ def assert_count(actual: float, expected: int, flag: str) -> None:
 
 
 def assert_metric_activity(
-    metrics: Iterable[dict[str, Any]], flag: str, *, allocation_key: str, suppressed: bool
+    metrics: Iterable[dict[str, Any]], flag: str, *, allocation_key: str | None, suppressed: bool
 ) -> None:
     # Agent series do not retain the source OTel aggregation temporality. Check
     # presence, not a sum that can count repeated cumulative exports twice.
@@ -44,7 +44,7 @@ def assert_metric_activity(
         for metric in metrics
         if metric.get("metric") == "feature_flag.evaluations"
         and f"feature_flag.key:{flag}" in metric.get("tags", [])
-        and f"feature_flag.result.allocation_key:{allocation_key}" in metric.get("tags", [])
+        and (allocation_key is None or f"feature_flag.result.allocation_key:{allocation_key}" in metric.get("tags", []))
         for point in metric.get("points", [])
     ]
     if suppressed:
diff --git a/utils/_context/_scenarios/__init__.py b/utils/_context/_scenarios/__init__.py
index f5222fedc..b2548b16c 100644
--- a/utils/_context/_scenarios/__init__.py
+++ b/utils/_context/_scenarios/__init__.py
@@ -62,10 +62,16 @@ _AI_GUARD_WEBLOG_ENV = {
     "DD_APP_KEY": "mock_app_key",
 }
 
-# Keep the opt-out scenario identical to the ordinary FFE scenario except for activation.
+# Shared by FEATURE_FLAGGING_AND_EXPERIMENTATION and the evaluation telemetry opt-out scenarios.
+# The opt-out scenarios also enable span enrichment, and the enabled one sets the opt-out option.
 _FFE_WEBLOG_ENV: dict[str, str | None] = {
     "DD_EXPERIMENTAL_FLAGGING_PROVIDER_ENABLED": "true",
     "DD_FEATURE_FLAGS_CONFIGURATION_SOURCE": "remote_config",
+    # set_provider() in Python blocks until we receive RC configuration for feature flags.
+    # But it is only sent after weblog successfully boots and tests start executing.
+    # Python's OpenFeature SDK does not have "set provider and don't wait", so we reduce
+    # the timeout here, so that the provider initialization fails fast, weblog boots, and
+    # provider recovers when we set RC configuration later.
     "DD_EXPERIMENTAL_FLAGGING_PROVIDER_INITIALIZATION_TIMEOUT_MS": "100",
     "DD_REMOTE_CONFIG_POLL_INTERVAL_SECONDS": "0.2",
     "DD_METRICS_OTEL_ENABLED": "true",
@@ -903,7 +909,6 @@ class _Scenarios:
         rc_api_enabled=True,
         # Allow final EVP batches to reach the backend after the weblog flushes and stops.
         agent_interface_timeout=15,
-        # Fail initial provider setup promptly; the fixture delivers RC after startup.
         weblog_env=_FFE_WEBLOG_ENV,
         doc="",
         scenario_groups=[scenario_groups.ffe],